package com.qzuser.sendmessagebasic.server;

import com.qzuser.sendmessagebasic.client.req.*;
import com.qzuser.sendmessagebasic.client.resp.RespPayload;
import com.qzuser.sendmessagebasic.client.resp.Response;
import com.qzuser.sendmessagebasic.client.resp.SubscribeEntity;
import com.qzuser.sendmessagebasic.common.exception.BrokerException;
import com.qzuser.sendmessagebasic.common.exception.ConsumeException;
import com.qzuser.sendmessagebasic.common.exception.ToolException;
import com.qzuser.sendmessagebasic.common.tool.BinaryTool;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Broker {
    private ServerSocket serverSocket;
    private final VirtualHost virtualHost = new VirtualHost("default");
    private final ConcurrentHashMap<String, Socket> session = new ConcurrentHashMap<>();
    private ExecutorService executor;
    private volatile boolean runnable = true;

    public Broker(int port) {
        try {
            serverSocket = new ServerSocket(port);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void start() {
        executor = Executors.newCachedThreadPool();
        System.out.println("[Broker] 服务器启动");
        while (runnable) {
            try {
                Socket client = serverSocket.accept();
                executor.submit(() -> {
                    connection(client);
                });
            } catch (SocketException e) {
                System.out.println("[Broker] 服务器停止");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private void connection(Socket client) {
        try (DataInputStream in = new DataInputStream(client.getInputStream());
             DataOutputStream out = new DataOutputStream(client.getOutputStream())) {
            while (true) {
                Request request = readRequest(in);
                Response response = countResponse(request, client);
                writeResponse(out, response);
            }
        } catch (EOFException | SocketException e) {
            System.out.println("[Broker] connection关闭 -> " + client.getInetAddress().toString() + ":" + client.getPort());
        } catch (IOException | BrokerException | ToolException e) {
            e.printStackTrace();
        } finally{
            try {
                client.close();
                clearClosedSessions(client);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private Request readRequest(DataInputStream in) throws IOException {
        Request request = new Request();
        request.setType(in.readInt());
        request.setLength(in.readInt());
        byte[] payload = new byte[request.getLength()];
        in.readFully(payload);
        request.setPayload(payload);
        return request;
    }

    private Response countResponse(Request request, Socket client) throws ToolException, BrokerException {
        ReqPayload reqPayload = (ReqPayload) BinaryTool.fromByteArray(request.getPayload());
        int type = request.getType();
        System.out.println("[Broker] 接收到客户端请求：\n\t" +
                "Request: {" + "\n\t\t" +
                "type: " + type + ",\n\t\t" +
                "length: " + request.getLength() + ",\n\t\t" +
                "rid: " + reqPayload.getRid() + ",\n\t\t" +
                "channelId: " + reqPayload.getChannelId() + "\n\t" +
                '}');
        boolean ok = true;
        if (type == 0x1) {
            session.put(reqPayload.getChannelId(), client);
            System.out.println("[Broker] channel创建成功：" + reqPayload.getChannelId());
        } else if (type == 0x2) {
            session.remove(reqPayload.getChannelId());
            System.out.println("[Broker] channel销毁成功：" + reqPayload.getChannelId());
        } else if (type == 0x3) {
            ExchangeEntity exchange = (ExchangeEntity) reqPayload;
            ok = virtualHost.exchangeDeclare(exchange.getName(), exchange.getType(), exchange.getDurable(), exchange.getAutoDelete(), exchange.getArguments());
        } else if (type == 0x4) {
            ExchangeEntity exchange = (ExchangeEntity) reqPayload;
            ok = virtualHost.exchangeDelete(exchange.getName());
        } else if (type == 0x5) {
            QueueEntity queue = (QueueEntity) reqPayload;
            ok = virtualHost.queueDeclare(queue.getName(), queue.getDurable(), queue.getExclusive(), queue.getAutoDelete(), queue.getArguments());
        } else if (type == 0x6) {
            QueueEntity queue = (QueueEntity) reqPayload;
            ok = virtualHost.queueDelete(queue.getName());
        } else if (type == 0x7) {
            BindEntity binding = (BindEntity) reqPayload;
            ok = virtualHost.queueBind(binding.getExchangeName(), binding.getQueueName(), binding.getBindingKey());
        } else if (type == 0x8) {
            BindEntity binding = (BindEntity) reqPayload;
            ok = virtualHost.queueUnbind(binding.getExchangeName(), binding.getQueueName());
        } else if (type == 0x9) {
            PublishEntity publish = (PublishEntity) reqPayload;
            ok = virtualHost.publish(publish.getExchangeName(), publish.getRoutingKey(), publish.getProperties(), publish.getBody());
        } else if (type == 0xa) {
            ConsumeEntity consume = (ConsumeEntity) reqPayload;
            ok = virtualHost.consume(consume.getConsumerTag(), consume.getQueueName(), consume.getAutoAck(), (tag, pro, body) -> {
                synchronized (consume) {
                    Socket socket = session.get(tag);
                    if (socket == null || socket.isClosed()) {
                        throw new BrokerException("[Broker] 订阅失败，当前客户端已经关闭");
                    }
                    SubscribeEntity subscribe = new SubscribeEntity();
                    subscribe.setRid("");
                    subscribe.setChannelId(tag);
                    subscribe.setOk(true);
                    subscribe.setConsumerTag(tag);
                    subscribe.setProperties(pro);
                    subscribe.setBody(body);
                    byte[] byteArray = BinaryTool.toByteArray(subscribe);
                    Response response = new Response();
                    response.setType(0xc);
                    response.setLength(byteArray.length);
                    response.setPayload(byteArray);
                    DataOutputStream out = new DataOutputStream(client.getOutputStream());
                    writeResponse(out, response);
                }
            });
        } else if (type == 0xb) {
            HandAckEntity ack = (HandAckEntity) reqPayload;
            ok = virtualHost.basicAck(ack.getQueueName(), ack.getMessageId());
        } else {
            throw new BrokerException("[Broker] 请求异常，未知请求类型：" + type);
        }
        RespPayload respPayload = new RespPayload();
        respPayload.setRid(reqPayload.getRid());
        respPayload.setChannelId(reqPayload.getChannelId());
        respPayload.setOk(ok);
        byte[] byteArray = BinaryTool.toByteArray(respPayload);
        Response response = new Response();
        response.setType(request.getType());
        response.setLength(byteArray.length);
        response.setPayload(byteArray);
        return response;
    }

    private void writeResponse(DataOutputStream out, Response response) throws IOException {
        out.writeInt(response.getType());
        out.writeInt(response.getLength());
        out.write(response.getPayload());
        out.flush();
    }

    private void clearClosedSessions(Socket client) {
        ArrayList<String> toDeleteChannel = new ArrayList<>();
        session.forEach((channel, socket) -> {
            if (socket == client) {
                toDeleteChannel.add(channel);
            }
        });
        System.out.println("[Broker] session清理完毕：" + toDeleteChannel);
        toDeleteChannel.forEach(session::remove);
    }

    public void stop() {
        try {
            runnable = false;
            executor.shutdownNow();
            serverSocket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
